package com.qisi.functions;

import com.googlecode.aviator.AviatorEvaluator;
import com.qisi.pojo.LogMessage;
import com.qisi.pojo.Rule;
import com.qisi.services.RulesService;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

/**
 * RulesService 周期性查询 MySQL 更新规则，筛选出明细 LogMessage
 */
@Deprecated
public class AlertMessageFilterFunction extends ProcessFunction<LogMessage, LogMessage> {

    private RulesService rulesService;

    private HashMap<Integer, Rule> rules;

    private long time = 0;
    private long period;

    public AlertMessageFilterFunction(long period) {
        this.period = period;
    }

    @Override
    public void open(Configuration parameters) throws SQLException {
        rulesService = new RulesService(parameters);
        rules = rulesService.getRules();
    }

    @Override
    public void processElement(LogMessage value, Context ctx, Collector<LogMessage> out) throws Exception {

        long current = ctx.timerService().currentProcessingTime();
        long period = current - time;
        if (period > this.period) {
            rulesService.incrementSyncRule(value.toString());
            time = current;
        }

        for (Map.Entry<Integer, Rule> entry : rules.entrySet()) {

            String ruleExp = entry.getValue().getExp();
            if((boolean) AviatorEvaluator.execute(ruleExp, value.toMap())){
                out.collect(value);
            }

        }
    }

    @Override
    public void close() throws Exception {
        super.close();
    }


}
